using Microsoft.Extensions.Logging;
using SeataNet.Core.Configuration;
using SeataNet.Core.Exceptions;
using SeataNet.Core.Protocol;
using SeataNet.Core.Common;
using System.Collections.Concurrent;
using System.Net;
using System.Text.Json;

namespace SeataNet.Core.RPC
{
    /// <summary>
    /// RPC服务器接口
    ///
    /// 定义一个嵌入式的HTTP RPC服务器，用于接收客户端的RPC请求。
    /// 服务器支持消息处理器的动态注册和注销。
    /// </summary>
    public interface IRpcServer
    {
        /// <summary>
        /// 异步启动RPC服务器
        ///
        /// 启动HTTP监听器，开始接收客户端请求。
        /// 服务器启动后会在后台线程中处理所有传入的HTTP请求。
        /// </summary>
        /// <param name="cancellationToken">取消令牌</param>
        /// <returns>启动操作的异步任务</returns>
        Task StartAsync(CancellationToken cancellationToken = default);

        /// <summary>
        /// 异步停止RPC服务器
        ///
        /// 停止HTTP监听，不再接收新的请求。
        /// 已接收但未处理的请求会被继续处理直到完成。
        /// </summary>
        /// <param name="cancellationToken">取消令牌</param>
        /// <returns>停止操作的异步任务</returns>
        Task StopAsync(CancellationToken cancellationToken = default);

        /// <summary>
        /// 获取服务器是否正在运行
        /// </summary>
        bool IsRunning { get; }

        /// <summary>
        /// 注册特定消息类型的处理器
        ///
        /// 当服务器收到该消息类型时，将调用注册的处理器进行处理。
        /// 同一消息类型只能注册一个处理器，新的处理器会覆盖旧的。
        /// </summary>
        /// <typeparam name="T">消息的类型</typeparam>
        /// <param name="handler">处理该消息类型的异步委托</param>
        void RegisterMessageHandler<T>(Func<T, Task<AbstractMessage>> handler) where T : AbstractMessage;

        /// <summary>
        /// 注销特定消息类型的处理器
        ///
        /// 移除该消息类型的处理器。
        /// 移除后，收到该类型消息时将返回错误。
        /// </summary>
        /// <typeparam name="T">消息的类型</typeparam>
        void UnregisterMessageHandler<T>() where T : AbstractMessage;
    }

    /// <summary>
    /// 默认RPC服务器实现
    ///
    /// 职责：
    /// 1. 使用.NET内置的HttpListener创建一个轻量级HTTP服务器
    /// 2. 接收和处理来自RPC客户端的请求
    /// 3. 支持心跳（心活）消息和业务消息的处理
    /// 4. 将请求分发到注册的消息处理器
    /// 5. 序列化和发送响应给客户端
    ///
    /// 架构特点：
    /// - 单线程接收请求，多线程处理请求
    /// - 后台任务处理HTTP请求流
    /// - 支持并发的客户端连接
    /// - 自动资源清理
    ///
    /// 消息路由：
    /// - /api/seata/heartbeat: 心跳请求
    /// - /api/seata/message: 业务消息请求
    /// - 其他路径返回404
    ///
    /// 线程安全：
    /// - 使用ConcurrentDictionary存储消息处理器
    /// - 使用volatile标志表示运行状态
    /// - 后台任务安全退出
    /// </summary>
    public class DefaultRpcServer : IRpcServer, IDisposable
    {
        private readonly SeataConfiguration _configuration;
        private readonly ILogger<DefaultRpcServer> _logger;
        private readonly ILoggerFactory _loggerFactory;
        // 消息类型 -> 处理器的映射表
        private readonly ConcurrentDictionary<Type, Func<AbstractMessage, Task<AbstractMessage>>> _messageHandlers;
        // 消息序列化器（JSON或二进制）
        private readonly IMessageSerializer _serializer;
        // HTTP监听器
        private readonly HttpListener _httpListener;
        // 取消令牌源，用于优雅关闭后台任务
        private readonly CancellationTokenSource _cancellationTokenSource;
        // 后台处理任务
        private readonly Task _processingTask;
        // 运行状态标志
        private volatile bool _isRunning = false;
        // 销毁状态标志
        private volatile bool _disposed = false;

        public bool IsRunning => _isRunning;

        public DefaultRpcServer(
            SeataConfiguration configuration,
            ILogger<DefaultRpcServer> logger,
            ILoggerFactory loggerFactory,
            IMessageSerializer? serializer = null)
        {
            _configuration = configuration;
            _logger = logger;
            _loggerFactory = loggerFactory;
            _messageHandlers = new ConcurrentDictionary<Type, Func<AbstractMessage, Task<AbstractMessage>>>();
            // 使用提供的序列化器或默认JSON序列化器
            _serializer = serializer ?? new JsonMessageSerializer(_loggerFactory.CreateLogger<JsonMessageSerializer>());
            _cancellationTokenSource = new CancellationTokenSource();

            // 解析监听地址（如 http://localhost:8091/）
            var listenAddress = ParseListenAddress(configuration.Transport.ListenAddress);
            _httpListener = new HttpListener();
            _httpListener.Prefixes.Add(listenAddress);

            // 启动后台处理任务，在后台线程持续接收和处理HTTP请求
            _processingTask = Task.Run(ProcessRequestsAsync, _cancellationTokenSource.Token);
        }

        /// <summary>
        /// 启动RPC服务器
        ///
        /// 流程：
        /// 1. 检查是否已销毁或已运行
        /// 2. 启动HttpListener开始接收请求
        /// 3. 标记服务器为运行状态
        /// 4. 后台任务将开始处理传入的HTTP请求
        /// </summary>
        /// <param name="cancellationToken">取消令牌</param>
        /// <returns>启动操作的异步任务</returns>
        public async Task StartAsync(CancellationToken cancellationToken = default)
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(nameof(DefaultRpcServer));
            }

            if (_isRunning)
            {
                return;
            }

            try
            {
                _httpListener.Start();
                _isRunning = true;
                _logger.LogInformation($"RPC服务器已启动，监听地址: {_configuration.Transport.ListenAddress}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "启动RPC服务器失败");
                throw new RpcException("启动RPC服务器失败", ex);
            }
        }

        /// <summary>
        /// 停止RPC服务器
        ///
        /// 流程：
        /// 1. 检查服务器是否正在运行
        /// 2. 停止HttpListener，不再接收新请求
        /// 3. 标记服务器为停止状态
        /// 4. 已接收的请求继续处理直到完成
        /// </summary>
        /// <param name="cancellationToken">取消令牌</param>
        /// <returns>停止操作的异步任务</returns>
        public async Task StopAsync(CancellationToken cancellationToken = default)
        {
            if (!_isRunning)
            {
                return;
            }

            try
            {
                _httpListener.Stop();
                _isRunning = false;
                _logger.LogInformation("RPC服务器已停止");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "停止RPC服务器失败");
                throw new RpcException("停止RPC服务器失败", ex);
            }
        }

        /// <summary>
        /// 注册消息处理器
        ///
        /// 注册一个异步委托来处理特定类型的消息。
        /// 当服务器收到该消息类型时，将调用此委托进行处理。
        ///
        /// 示例：
        /// server.RegisterMessageHandler<GlobalBeginRequest>(async msg => {
        ///     // 处理消息
        ///     return new GlobalBeginResponse { ... };
        /// });
        ///
        /// 线程安全：使用ConcurrentDictionary，支持并发的消息处理
        /// </summary>
        /// <typeparam name="T">消息类型</typeparam>
        /// <param name="handler">处理该消息的异步委托</param>
        public void RegisterMessageHandler<T>(Func<T, Task<AbstractMessage>> handler) where T : AbstractMessage
        {
            if (handler == null)
            {
                throw new ArgumentNullException(nameof(handler));
            }

            var messageType = typeof(T);
            _messageHandlers[messageType] = async (message) =>
            {
                if (message is T typedMessage)
                {
                    return await handler(typedMessage);
                }

                throw new InvalidOperationException($"消息类型不匹配: 期望 {messageType.Name}, 实际 {message.GetType().Name}");
            };

            _logger.LogDebug($"注册消息处理器: {messageType.Name}");
        }

        /// <summary>
        /// 注销消息处理器
        ///
        /// 移除已注册的消息处理器。
        /// 移除后，收到该类型的消息将返回错误。
        /// </summary>
        /// <typeparam name="T">消息类型</typeparam>
        public void UnregisterMessageHandler<T>() where T : AbstractMessage
        {
            var messageType = typeof(T);
            _messageHandlers.TryRemove(messageType, out _);
            _logger.LogDebug($"取消注册消息处理器: {messageType.Name}");
        }

        /// <summary>
        /// 后台请求处理任务
        ///
        /// 这是一个无限循环，不断从HttpListener中提取HTTP上下文并处理。
        /// 每个HTTP请求都会在新的任务中异步处理，实现并发处理。
        ///
        /// 流程：
        /// 1. 使用GetContextAsync()等待HTTP请求
        /// 2. 为每个请求创建新任务进行处理
        /// 3. 当取消令牌被触发时优雅退出
        /// 4. ObjectDisposedException表示HttpListener已关闭
        /// </summary>
        private async Task ProcessRequestsAsync()
        {
            try
            {
                while (!_cancellationTokenSource.Token.IsCancellationRequested)
                {
                    try
                    {
                        // 等待下一个HTTP请求
                        var context = await _httpListener.GetContextAsync();
                        // 在新任务中异步处理请求，不阻塞后续请求的接收
                        _ = Task.Run(async () => await ProcessRequestAsync(context), _cancellationTokenSource.Token);
                    }
                    catch (ObjectDisposedException)
                    {
                        // HttpListener已关闭，正常退出
                        break;
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "处理请求时发生错误");
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理请求任务异常退出");
            }
        }

        /// <summary>
        /// 处理单个HTTP请求
        ///
        /// 流程：
        /// 1. 检查请求方法（只支持POST）
        /// 2. 根据URL路径路由请求：
        ///    - /heartbeat -> 心跳处理
        ///    - /message -> 业务消息处理
        ///    - 其他 -> 404错误
        /// 3. 读取请求体数据
        /// 4. 反序列化消息
        /// 5. 调用相应的处理器
        /// 6. 序列化响应并发送
        ///
        /// 错误处理：
        /// - 405: 方法不允许
        /// - 404: 路径不存在
        /// - 500: 服务器内部错误
        /// </summary>
        /// <param name="context">HTTP上下文</param>
        private async Task ProcessRequestAsync(HttpListenerContext context)
        {
            try
            {
                var request = context.Request;
                var response = context.Response;

                // 检查请求方法：只接受POST请求
                if (request.HttpMethod != "POST")
                {
                    response.StatusCode = 405; // Method Not Allowed
                    response.Close();
                    return;
                }

                // 根据URL路径路由请求
                if (request.Url?.AbsolutePath.EndsWith("/heartbeat") == true)
                {
                    // 心跳请求：保活连接
                    await ProcessHeartbeatRequestAsync(context);
                    return;
                }

                // 业务消息请求
                if (request.Url?.AbsolutePath.EndsWith("/message") == true)
                {
                    await ProcessMessageRequestAsync(context);
                    return;
                }

                // 未知请求路径
                response.StatusCode = 404; // Not Found
                response.Close();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理HTTP请求时发生错误");

                try
                {
                    var response = context.Response;
                    response.StatusCode = 500; // Internal Server Error
                    response.Close();
                }
                catch
                {
                    // 忽略关闭响应时的错误
                }
            }
        }

        /// <summary>
        /// 处理心跳请求
        ///
        /// 心跳用于保活连接。
        /// 处理流程：
        /// 1. 读取心跳请求数据
        /// 2. 反序列化为HeartbeatRequest对象
        /// 3. 构造HeartbeatResponse响应
        /// 4. 序列化响应并发送
        ///
        /// 响应包含：
        /// - MessageId: 消息ID（可选）
        /// - ResultCode: 结果代码（Success表示连接正常）
        /// </summary>
        /// <param name="context">HTTP上下文</param>
        private async Task ProcessHeartbeatRequestAsync(HttpListenerContext context)
        {
            var request = context.Request;
            var response = context.Response;

            try
            {
                // 读取请求数据
                var requestData = await ReadRequestDataAsync(request);
                var heartbeatRequest = (HeartbeatRequest)_serializer.Deserialize(requestData);

                // 构造心跳响应
                var heartbeatResponse = new HeartbeatResponse
                {
                    MessageId = heartbeatRequest.MessageId,
                    ResultCode = ResultCode.Success
                };

                // 发送响应
                var responseData = _serializer.Serialize(heartbeatResponse);
                await WriteResponseDataAsync(response, responseData);

                _logger.LogDebug("处理心跳请求成功");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理心跳请求失败");

                response.StatusCode = 500; // Internal Server Error
                response.Close();
            }
        }

        /// <summary>
        /// 处理业务消息请求
        ///
        /// 处理流程：
        /// 1. 读取请求数据
        /// 2. 反序列化为消息对象
        /// 3. 检查是否是单向消息（X-Oneway头）
        /// 4. 调用对应的消息处理器
        /// 5. 如果是双向消息，序列化和发送响应
        ///
        /// 单向消息：
        /// - 不需要等待处理完成
        /// - 立即返回200 OK响应
        /// - 用于通知、日志等场景
        ///
        /// 双向消息：
        /// - 等待处理完成
        /// - 返回处理结果
        /// - 用于需要响应的RPC调用
        /// </summary>
        /// <param name="context">HTTP上下文</param>
        private async Task ProcessMessageRequestAsync(HttpListenerContext context)
        {
            var request = context.Request;
            var response = context.Response;

            try
            {
                // 读取请求数据
                var requestData = await ReadRequestDataAsync(request);
                var message = _serializer.Deserialize(requestData);

                // 检查是否是单向消息（通过HTTP头X-Oneway标记）
                var isOneway = request.Headers["X-Oneway"] == "true";

                if (isOneway)
                {
                    // 单向消息：不等待处理完成，立即返回
                    await ProcessMessageAsync(message);
                    response.StatusCode = 200; // OK
                    response.Close();
                    return;
                }

                // 双向消息：等待处理完成并返回响应
                var responseMessage = await ProcessMessageAsync(message);

                // 发送响应
                var responseData = _serializer.Serialize(responseMessage);
                await WriteResponseDataAsync(response, responseData);

                _logger.LogDebug($"处理消息请求成功: {message.GetType().Name}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理消息请求失败");

                response.StatusCode = 500; // Internal Server Error
                response.Close();
            }
        }

        /// <summary>
        /// 调用消息处理器处理消息
        ///
        /// 流程：
        /// 1. 根据消息类型查找注册的处理器
        /// 2. 调用处理器处理消息
        /// 3. 返回处理结果
        ///
        /// 异常处理：
        /// - 如果未找到处理器，抛出RpcException
        /// </summary>
        /// <param name="message">待处理的消息</param>
        /// <returns>处理结果消息</returns>
        private async Task<AbstractMessage> ProcessMessageAsync(AbstractMessage message)
        {
            var messageType = message.GetType();

            if (!_messageHandlers.TryGetValue(messageType, out var handler))
            {
                throw new RpcException($"未找到消息处理器: {messageType.Name}");
            }

            return await handler(message);
        }

        /// <summary>
        /// 读取HTTP请求体数据
        ///
        /// 从HttpListenerRequest的InputStream中读取所有字节数据。
        /// 适用于读取POST请求体。
        /// </summary>
        /// <param name="request">HTTP请求对象</param>
        /// <returns>请求体字节数据</returns>
        private async Task<byte[]> ReadRequestDataAsync(HttpListenerRequest request)
        {
            using var memoryStream = new MemoryStream();
            await request.InputStream.CopyToAsync(memoryStream);
            return memoryStream.ToArray();
        }

        /// <summary>
        /// 写入HTTP响应数据
        ///
        /// 将字节数据写入HttpListenerResponse的OutputStream。
        /// 设置响应的内容类型和长度。
        /// </summary>
        /// <param name="response">HTTP响应对象</param>
        /// <param name="data">响应体字节数据</param>
        private async Task WriteResponseDataAsync(HttpListenerResponse response, byte[] data)
        {
            response.ContentType = "application/octet-stream";
            response.ContentLength64 = data.Length;

            await response.OutputStream.WriteAsync(data, 0, data.Length);
            response.Close();
        }

        /// <summary>
        /// 解析并规范化监听地址
        ///
        /// 输入示例：
        /// - "localhost:8091" -> "http://localhost:8091/"
        /// - "http://0.0.0.0:8091" -> "http://0.0.0.0:8091/"
        /// - ":8091" -> "http://localhost:8091/"
        ///
        /// 处理流程：
        /// 1. 如果为空，使用默认值
        /// 2. 如果没有协议前缀，添加"http://"
        /// 3. 如果没有端口，添加默认端口8091
        /// 4. 确保以"/"结尾（HttpListener要求）
        /// </summary>
        /// <param name="listenAddress">监听地址字符串</param>
        /// <returns>规范化后的监听地址</returns>
        private string ParseListenAddress(string listenAddress)
        {
            if (string.IsNullOrWhiteSpace(listenAddress))
            {
                return "http://localhost:8091/";
            }

            var address = listenAddress.Trim();

            // 如果没有协议前缀，添加http://
            if (!address.StartsWith("http://") && !address.StartsWith("https://"))
            {
                address = "http://" + address;
            }

            // 如果没有端口，添加默认端口8091
            if (!address.Contains(":") || address.EndsWith(":"))
            {
                address = address.TrimEnd(':') + ":8091";
            }

            // 确保以/结尾
            if (!address.EndsWith("/"))
            {
                address += "/";
            }

            return address;
        }

        /// <summary>
        /// 释放资源
        ///
        /// 流程：
        /// 1. 检查是否已销毁
        /// 2. 异步停止服务器（如果正在运行）
        /// 3. 取消后台处理任务
        /// 4. 关闭HttpListener
        /// 5. 清空消息处理器字典
        /// 6. 标记为已销毁
        /// </summary>
        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;

            try
            {
                StopAsync().GetAwaiter().GetResult();
            }
            catch
            {
                // 忽略停止时的错误
            }

            _cancellationTokenSource?.Cancel();
            _cancellationTokenSource?.Dispose();

            _httpListener?.Stop();
            _httpListener?.Close();

            _messageHandlers.Clear();
        }
    }
}